package com.aspire.kafkaconsumer.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * kafka消费者工厂类
 *
 * @author kingSon
 *
 * @version 1.0.0
 *
 */
@EnableKafka
@Configuration
public class KafkaConsumerConfiguration {

    @Autowired
    private KafkaConf kafkaConf;

    /**
     * 设置消费者监听内容工厂
     *
     * @return KafkaListenerContainerFactory
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> userKafkaListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = null;
        try {
            factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(this.consumerFactory());
            factory.setConcurrency(3);
            factory.getContainerProperties().setPollTimeout(3000);
        } catch (final Exception e) {
        }
        return factory;
    }

    /**
     * 设置消费者连接工厂
     *
     * @return ConsumerFactory<String, String>
     */
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(this.consumerConfigs());
    }

    /**
     * 设置消费者连接配置
     *
     * @return Map<String, Object>
     */
    private Map<String, Object> consumerConfigs() {
        final Map<String, Object> propsMap = new ConcurrentHashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConf.getBootstrapServers());
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, this.kafkaConf.isEnableAutoCommit());
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, this.kafkaConf.getAutoCommitIntervalMs());
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, this.kafkaConf.getSessionTimeoutMs());
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, this.kafkaConf.getGroupId());
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, this.kafkaConf.getAutoOffsetReset());
        return propsMap;
    }
}
